package com.ccnu.demojava.qtpmessage;

//import scala.tools.nsc.Global;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.ConcurrentLinkedQueue;

public class SocketClient {
    private ConcurrentLinkedQueue<QtpMessage> sendQueue = new ConcurrentLinkedQueue<>();
    private ConcurrentLinkedQueue<QtpMessage> recvQueue = new ConcurrentLinkedQueue<>();
    private Socket client = null;
    private InetSocketAddress serverAddress = null;
    boolean stop = true;
    // 读写线程
    private Thread sendThread = new Thread(new SendThread());
    private Thread recvThread = new Thread(new RecvThread());

    private class SendThread implements Runnable {
        @Override
        public void run() {
            while (!stop) {
                try {
                    if (client == null) {
                        Thread.sleep(100);
                        continue;
                    }
                    // 如果客户端没有连接，并且重连失败
                    if (!client.isConnected() && client.isClosed() && !reconnect()) {
                        Thread.sleep(500);
                        continue;
                    }
                    QtpMessage message = sendQueue.poll();
                    if (message != null && !client.isOutputShutdown()) {
                        client.getOutputStream().write(message.encode());
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private class RecvThread implements Runnable {
        private static final int MAX_BUFFER = 32*1024;
        private byte[] buffer = new byte[MAX_BUFFER];
        @Override
        public void run() {
            byte[] bufT = new byte[MAX_BUFFER]; // 临时变量，用于缓冲
            int offset = 0;
            while (!stop) {
                try {
                    if (client == null) {
                        Thread.sleep(100);
                        continue;
                    }
                    // 如果客户端没有连接，并且重连失败
                    if (!client.isConnected() && client.isClosed() && !reconnect()) {
                        Thread.sleep(500);
                        continue;
                    }
                    // 如果输入关闭,则继续
                    if (client.isInputShutdown()) {
                        continue;
                    }
                    // 一次性读取最大的数量
                    int read = client.getInputStream().read(buffer, offset, MAX_BUFFER-offset);
                    // TODO 如果流已经结束了，继续等待. 如果30秒都没有继续发送流，则丢失数据。重新开始读取
                    if (read == -1) {
                        // TODO 此处应该计时
                    } else {
                        offset = offset + read;
                        // 如果小于最小要求，则继续读流
                        if (offset < QtpMessage.MsgHead.SIZE) {
                            continue;
                        }
                        QtpMessage.MsgHead msgHead = new QtpMessage.MsgHead();
                        msgHead.fromBytes(buffer);
                        int msgLen = (int) (QtpMessage.MsgHead.SIZE + msgHead.getOptslen() + msgHead.getDatalen());
                        if (offset < msgLen) {
                            continue;
                        }
                        // 如果完整接收到了报文，则开始处理
                        byte []dataBuf = null;
                        byte []optBuf = null;
                        if (msgHead.getOptslen() > 0) {
                            optBuf = new byte[msgHead.getOptslen()];
                            System.arraycopy(buffer, QtpMessage.MsgHead.SIZE, optBuf, 0, msgHead.getOptslen());
                        }
                        if (msgHead.getDatalen() > 0) {
                            dataBuf = new byte[(int) msgHead.getDatalen()];
                            System.arraycopy(buffer, QtpMessage.MsgHead.SIZE + msgHead.getOptslen(), dataBuf, 0, (int) msgHead.getDatalen());
                        }
                        QtpMessage msgRecv = new QtpMessage(msgHead, dataBuf, optBuf);
                        recvQueue.add(msgRecv);

                        // 重置缓冲区大小，并且处理粘包
                        offset = offset - msgLen;
                        if (offset < 0) {
                            offset = 0;
                        }
                        if (offset > 0) {
                            // 多余的数据要前移
                            int remainLen = MAX_BUFFER - msgLen;
                            System.arraycopy(buffer, msgLen, bufT, 0, remainLen);
                            System.arraycopy(bufT, 0, buffer, 0, remainLen);
                        }
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * 连接到指定的ip
     *
     * @param ip
     * @param port
     * @return
     */
    public boolean connect(String ip, int port) {
        try {
            if (client != null && client.isConnected()) {
                client.close();
            }
            client = new Socket();
            serverAddress = new InetSocketAddress(ip, port);
            // 连接服务器，超时30秒。允许服务器没有连接上的情况
            client.connect(serverAddress, 30);
            stop = false;
            sendThread.start();
            recvThread.start();
        } catch (IOException e) {
            System.out.println(e.getMessage());
            return false;
        }
        return true;
    }

    /**
     * 重连
     * @return
     */
    public boolean reconnect() {
        // 重连的时候应该加锁
        if (client == null || serverAddress == null) {
            return false;
        }
        try {
            client.connect(serverAddress, 30);
        } catch (IOException e) {
            System.out.println(e.getMessage());
            return false;
        }
        return true;
    }

    public boolean isConnected() {
        return client!=null && client.isConnected();
    }

    public boolean close() {
        this.stop = true;
        if (client != null) {
            try {
                client.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return true;
    }

    /**
     * 发送信息
     *
     * @param message
     * @return
     */
    public boolean send(QtpMessage message) {
        sendQueue.add(message);
        return true;
    }

    /**
     * 读取信息
     *
     * @return
     */
    public QtpMessage recv() {
        return recvQueue.poll();
    }
}
